package com.hrt.iceberg

import java.util

import org.apache.hadoop.conf.Configuration
import org.apache.iceberg.Table
import org.apache.iceberg.actions.Actions
import org.apache.iceberg.catalog.TableIdentifier
import org.apache.iceberg.hadoop.HadoopCatalog
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap
import org.apache.spark.sql.SparkSession

/**
  * 使用Hive Catalog管理iceberg表，默认数据存储在Hive 对应的datawarehouse目录下
  * 在Hive中会自动创建对应的iceberg 外表，但是目前在Hive中不能查询这个表中数据。
  */
object SparkOperateIcebergWithHiveCatalog {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local").appName("SparkOperateIceberg")
      //指定hive catalog, catalog名称为hive_prod
      .config("spark.sql.catalog.hive_prod", "org.apache.iceberg.spark.SparkCatalog")
      .config("spark.sql.catalog.hive_prod.type", "hive")
      .config("spark.sql.catalog.hive_prod.uri", "thrift://node1:9083")
      .config("iceberg.engine.hive.enabled", "true")
      //指定hadoop catalog，catalog名称为hadoop_prod
      .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")
      .config("spark.sql.catalog.hadoop_prod.type", "hadoop")
      .config("spark.sql.catalog.hadoop_prod.warehouse", "hdfs://mycluster/sparkoperateiceberg")
      .getOrCreate()

    //创建表 ，hive_pord：指定catalog名称。default:指定Hive中存在的库。test：创建的iceberg表名。
    spark.sql(
          """
            | create table if not exists hive_prod.default.test(id int,name string,age int) using iceberg
          """.stripMargin)

    //插入数据
    spark.sql(
      """
        |insert into hive_prod.default.test values (1,"zs",18),(2,"ls",19),(3,"ww",20)
      """.stripMargin)

    //查询数据
//    spark.sql(
//      """
//        |select * from hive_prod.default.test
//      """.stripMargin).show()

    //删除表，删除表对应的数据不会被删除
//    spark.sql(
//      """
//        |drop table hive_prod.default.test
//      """.stripMargin)

  }

}
